package cn.doitedu.rtmk.demo9;

import cn.doitedu.rtmk.beans.UserEvent;
import cn.doitedu.rtmk.interfaces.IRuleCalculatorV3;
import cn.doitedu.rtmk.utils.RuleHitResultBuilder;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.RoaringBitmap;
import redis.clients.jedis.Jedis;


/**
 * 本规则运算模型可以支持类似如下的规则
 *   静态画像条件： 月平均消费额 > 200 &&  age: [20,30] && gender=male
 *   动态实时画像条件： 规则上线后, 发生过 A行为 >=3次 (A行为: event_id = A, propName='',propValue = '')
 *   触发事件条件 : event_id = B, propertyName = p1 , propertyValue = v1
 {
 "rule_id":3,
 "rule_model_id":2,
 "static_profile_conditions":[
 {"tag_name":"TAG0101","compare":">","tag_value":[200]},
 {"tag_name":"TAG0102","compare":"bt","tag_value":[20,30]},
 {"tag_name":"TAG0103","compare":"=","tag_value":["male"]}
 ],
 "dynamic_profile_conditions":{
 "condition_id":"condition_01",
 "event_id":"add_cart",
 "event_property_name":"item_id",
 "event_property_value":"item001",
 "window_start":"2023-04-01",
 "window_end":"2023-04-30"
 "minCount":3
 },
 "trigger_condition":{
 "event_id":"add_cart",
 "event_property_name":"item_id",
 "event_property_value":"item001"
 }

 }

*/
public class RuleModel3Calculator implements IRuleCalculatorV3 {

    RoaringBitmap preSelectUsersBitmap;
    JSONObject ruleInstanceParams;

    MapState<Long, Integer> countState;
    //HashMap<Long, Integer> countState;

    Integer rule_id;
    Integer rule_model_id;
    Jedis jedis;


    @Override
    public void open(RoaringBitmap preSelectUsersBitmap, JSONObject ruleInstanceParams, RuntimeContext runtimeContext) {
        this.preSelectUsersBitmap = preSelectUsersBitmap;
        this.ruleInstanceParams = ruleInstanceParams;

        // 取出规则的id
        rule_id = ruleInstanceParams.getInteger("rule_id");
        rule_model_id = ruleInstanceParams.getInteger("rule_model_id");


        // 声明一个MapState,用来记录各个用户 本规则的动态画像行为 发生的次数
        countState = runtimeContext.getMapState(new MapStateDescriptor<Long, Integer>("st", Long.class, Integer.class));
        //countState = new HashMap<>();

        // 构造改规则模型所需要的 redis客户端
        jedis = new Jedis("doitedu", 6379);

    }

    @Override
    public void calc(UserEvent userEvent, Collector<JSONObject> out) throws Exception {

        //
        Integer integer1 = countState.get(userEvent.getUser_id());
        System.out.println("进入calc的时候，取一次状态数据: " + integer1);

        if(preSelectUsersBitmap.contains((int)userEvent.getUser_id())) {
            /**
             * 从参数json中获取 动态画像条件的各个参数
             */
            JSONObject dynamic_profile_conditions = ruleInstanceParams.getJSONObject("dynamic_profile_conditions");
            String dynamic_event_id = dynamic_profile_conditions.getString("event_id");
            String dynamic_event_property_name = dynamic_profile_conditions.getString("event_property_name");
            String dynamic_event_property_value = dynamic_profile_conditions.getString("event_property_value");

            // 本条件的标识：条件id
            String condition_id = dynamic_profile_conditions.getString("condition_id");


            int dynamic_minCount = dynamic_profile_conditions.getIntValue("minCount");


            /**
             * 从参数json中获取 触发条件的各个参数
             */
            JSONObject trigger_condition = ruleInstanceParams.getJSONObject("trigger_condition");
            String trigger_event_id = trigger_condition.getString("event_id");
            String trigger_event_property_name = trigger_condition.getString("event_property_name");
            String trigger_event_property_value = trigger_condition.getString("event_property_value");


            /**
             * 判断 当前的用户行为，是否是 本规则动态画像条件所关心的行为
             */
            if (userEvent.getEvent_id().equals(dynamic_event_id) && userEvent.getProperties().get(dynamic_event_property_name).equals(dynamic_event_property_value)) {
                // 进行实时的画像标签计算
                Integer oldValue = countState.get(userEvent.getUser_id());

                // 如果oldValue为空，则需要从redis中获取该条件该用户的初始值
                // 大key ==>  规则id:条件id
                //  value  ==> Hash<user_id,initValue>
                if(oldValue == null) {  // 如果从状态中获取统计值为null，则从 redis中取获取 画像条件的初始值
                    String redisInitValueStr = jedis.hget(rule_id + ":" + condition_id, userEvent.getUser_id() + "");
					System.out.println(userEvent.getUser_id()+"的redis初始值为: " +redisInitValueStr );
                    oldValue = redisInitValueStr == null ? 0 : Integer.parseInt(redisInitValueStr);
                    System.out.println(userEvent.getUser_id()+"的oldValue被赋值为: " +oldValue );
                }

                // 更新状态中的统计值
                countState.put(userEvent.getUser_id(), oldValue+1);
                Integer integer = countState.get(userEvent.getUser_id());
                System.out.println("刚放进去就马上取出来: " + integer);
            }

            /**
             * 判断 当前的用户行为，是否是 本规则触发条件所要求的行为
             */
            if (userEvent.getEvent_id().equals(trigger_event_id) && userEvent.getProperties().get(trigger_event_property_name).equals(trigger_event_property_value)) {
                // 查询该用户是否已经满足各类受众条件
                Integer dynamicValue = countState.get(userEvent.getUser_id());
                System.out.println(userEvent.getUser_id()+"触发了规则,此时从状态中取到的条件值为:"+dynamicValue+"要求的minValue="+dynamic_minCount);

                if(dynamicValue != null && dynamicValue >= dynamic_minCount){
                    // 构造一条消息输出
                    JSONObject message = new RuleHitResultBuilder()
                            .setHitRuleModel(rule_model_id)
                            .setHitRule(rule_id)
                            .setHitTime(userEvent.getEvent_time())
                            .setUser(userEvent.getUser_id())
                            .build();

                    out.collect(message);
                }
            }

        }

    }
}
